package qiyu.live.living.provider.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import jakarta.annotation.Resource;
import org.idea.qiyu.live.framework.redis.starter.key.LivingRoomProviderCacheKeyBuilder;
import org.qiyu.live.common.interfaces.enums.CommonStatusEnum;
import org.qiyu.live.common.interfaces.utils.ConvertBeanUtils;
import org.qiyu.live.common.interfaces.vo.PageWrapper;
import org.qiyu.live.web.starter.context.QiyuRequestContext;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.qiyu.live.im.core.server.dto.ImOfflineDTO;
import org.qiyu.live.im.core.server.dto.ImOnlineDTO;
import qiyu.live.living.dto.req.LivingRoomReqDTO;
import qiyu.live.living.dto.req.OnlinePkReqDTO;
import qiyu.live.living.dto.resp.LivingRoomRespDTO;
import qiyu.live.living.provider.dao.mapper.ILivingRoomMapper;
import qiyu.live.living.provider.dao.mapper.ILivingRoomRecordMapper;
import qiyu.live.living.provider.dao.po.LivingRoomPO;
import qiyu.live.living.provider.dao.po.LivingRoomRecordPO;
import qiyu.live.living.provider.service.ILivingRoomService;

import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: sfx
 * Date: 2023-10-24
 * Time: 19:57
 */
@Service
public class LivingRoomServiceImpl implements ILivingRoomService {

    @Resource
    private ILivingRoomMapper livingRoomMapper;

    @Resource
    private ILivingRoomRecordMapper livingRoomRecordMapper;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private LivingRoomProviderCacheKeyBuilder cacheBuilder;

    /**
     * 查询直播列表
     *
     * @param livingRoomReqDTO dto
     * @return 返回直播列表
     */
    @Override
    public PageWrapper<LivingRoomRespDTO> list(LivingRoomReqDTO livingRoomReqDTO) {
        // 1. 先去redis中查询
        String key = cacheBuilder.buildLivingRoomList(livingRoomReqDTO.getType());
        Integer page = livingRoomReqDTO.getPage();
        Integer pageSize = livingRoomReqDTO.getPageSize();
        Long size = redisTemplate.opsForList().size(key);
        List<Object> range = redisTemplate.opsForList().range(key, (long) (page - 1) * pageSize, (long) page * pageSize);
        // 2. redis中为空,直接返回
        PageWrapper<LivingRoomRespDTO> pageWrapper = new PageWrapper<>();
        if (CollectionUtils.isEmpty(range)) {
            pageWrapper.setList(Collections.emptyList());
            pageWrapper.setHasNext(false);
        } else {
            // 3. redis不为空,使用list的range方法进行分页返回
            pageWrapper.setList(ConvertBeanUtils.convertList(range, LivingRoomRespDTO.class));
            pageWrapper.setHasNext((long) page * pageSize < size);
        }
        return pageWrapper;
    }

    @Override
    public LivingRoomRespDTO queryByRoomId(Integer roomId) {
        // 1. 先查询redis
        String key = cacheBuilder.buildLivingRoomObj(roomId);
        LivingRoomRespDTO queryResult = (LivingRoomRespDTO) redisTemplate.opsForValue().get(key);
        if (queryResult != null) {
            // 2. 防止缓存击穿这里加一个空值缓存
            if (queryResult.getId() == null) {
                return null;
            }
            return queryResult;
        }
        // 3. redis中不存在,查询数据库,缓存到redis中
        LambdaQueryWrapper<LivingRoomPO> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(LivingRoomPO::getStatus, CommonStatusEnum.VALID.getCode());
        queryWrapper.eq(LivingRoomPO::getId, roomId);
        queryWrapper.last("limit 1");
        LivingRoomRespDTO convert = ConvertBeanUtils.convert
                (livingRoomMapper.selectOne(queryWrapper), LivingRoomRespDTO.class);
        // 4. 数据库不存在,空值缓存到redis中
        if (convert == null) {
            redisTemplate.opsForValue().set(key, new LivingRoomRespDTO(), 5, TimeUnit.MINUTES);
        } else {
            redisTemplate.opsForValue().set(key, convert, 30, TimeUnit.MINUTES);
        }
        return convert;
    }

    @Override
    public List<LivingRoomRespDTO> queryAllLivingRoom(Integer type) {
        LambdaQueryWrapper<LivingRoomPO> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(LivingRoomPO::getType, type);
        queryWrapper.eq(LivingRoomPO::getStatus, CommonStatusEnum.VALID.getCode());
        queryWrapper.last("limit 1000");
        return ConvertBeanUtils.convertList(livingRoomMapper.selectList(queryWrapper), LivingRoomRespDTO.class);
    }

    @Override
    public Integer startLivingRoom(LivingRoomReqDTO livingRoomReqDTO) {
        LivingRoomPO convert = ConvertBeanUtils.convert(livingRoomReqDTO, LivingRoomPO.class);
        convert.setStatus(CommonStatusEnum.VALID.getCode());
        // 记录一下开播时间
        convert.setStartTime(new Date());
        livingRoomMapper.insert(convert);
        String key = cacheBuilder.buildLivingRoomObj(convert.getId());
        redisTemplate.delete(key);
        return convert.getId();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean closeLivingRoom(LivingRoomReqDTO livingRoomReqDTO) {
        LivingRoomPO livingRoomPO = livingRoomMapper.selectById(livingRoomReqDTO.getRoomId());
        if (livingRoomPO == null) {
            return false;
        }
        // 也必须是同一个主播才能够关闭直播
        Long userId = QiyuRequestContext.getUserId();
        if (!userId.equals(livingRoomPO.getAnchorId())) {
            return false;
        }
        LivingRoomRecordPO livingRoomRecordPO = ConvertBeanUtils.convert(livingRoomPO, LivingRoomRecordPO.class);
        // 设置关闭直播的时间
        livingRoomRecordPO.setEndTime(new Date());
        livingRoomRecordPO.setStatus(CommonStatusEnum.NO_VALID.getCode());
        // 添加到record记录里面
        livingRoomRecordMapper.insert(livingRoomRecordPO);
        // 删除livingRoom记录表中的记录
        livingRoomMapper.deleteById(livingRoomRecordPO.getId());
        String key = cacheBuilder.buildLivingRoomObj(livingRoomRecordPO.getId());
        redisTemplate.delete(key);
        return true;
    }

    @Override
    public void userOnlineHandler(ImOnlineDTO imOnlineDTO) {
        Integer appId = imOnlineDTO.getAppId();
        Long userId = imOnlineDTO.getUserId();
        Integer roomId = imOnlineDTO.getRoomId();
        String key = cacheBuilder.buildLivingRoomOnlineSet(appId, roomId);
        // key:roomId val:userId  -->这个直播间的在线用户,userId与roomId进行关联
        redisTemplate.opsForSet().add(key, userId);
    }

    @Override
    public void userOffLineHandler(ImOfflineDTO imOfflineDTO) {
        Integer appId = imOfflineDTO.getAppId();
        Long userId = imOfflineDTO.getUserId();
        Integer roomId = imOfflineDTO.getRoomId();
        String key = cacheBuilder.buildLivingRoomOnlineSet(appId, roomId);
        // 将这个userId从set集合中移除->用户已经下线了
        redisTemplate.opsForSet().remove(key, userId);
        // 这里监听用户下线,我们也要把pkObjId从redis中进行移除
        OnlinePkReqDTO onlinePkReqDTO = new OnlinePkReqDTO();
        onlinePkReqDTO.setRoomId(roomId);
        this.offlinePk(onlinePkReqDTO);
    }

    /**
     * 根据roomId查询整个房间的在线用户userId
     *
     * @param livingRoomReqDTO 直播间
     * @return 根据roomId查询整个房间的在线用户userId
     */
    @Override
    public List<Long> queryUserIdsByRoomId(LivingRoomReqDTO livingRoomReqDTO) {
        // 根据roomId批量的查询出userId,我们不能直接
        Integer appId = livingRoomReqDTO.getAppId();
        Integer roomId = livingRoomReqDTO.getRoomId();
        String key = cacheBuilder.buildLivingRoomOnlineSet(appId, roomId);
        // 使用scan命令进行扫描,100个100个进行批量扫描
        Cursor<Object> cursor = redisTemplate.opsForSet().scan(key, ScanOptions.scanOptions().count(100).build());
        List<Long> userIds = new ArrayList<>();
        while (cursor.hasNext()) {
            Integer userId = (Integer) cursor.next();
            userIds.add(Long.valueOf(userId));
        }
        return userIds;
    }

    @Override
    public boolean onlinePk(OnlinePkReqDTO onlinePkReqDTO) {
        // 当用户点击直播pk的时候,就代表想要与主播进行连线,我们这里需要记录一下连线的userId
        String cacheKey = cacheBuilder.buildLivingOnlinePkCacheKey(onlinePkReqDTO.getRoomId());
//        redisTemplate.opsForValue().set(cacheKey, Objects.requireNonNull(QiyuRequestContext.getUserId()));
//        return true;
        // 还有一点我们需要注意,如果有多个人点击pk按钮,那么只会有一个人连线成功,这里会使用set nx  --> 用户下线的时候我们就需要删除
        return Boolean.TRUE.equals(redisTemplate.opsForValue().setIfAbsent(cacheKey, Objects.requireNonNull(QiyuRequestContext.getUserId()), 30, TimeUnit.HOURS));
    }

    @Override
    public boolean offlinePk(OnlinePkReqDTO onlinePkReqDTO) {
        // 当用户下线的时候把对应的pk用户移除即可
        String cacheKey = cacheBuilder.buildLivingOnlinePkCacheKey(onlinePkReqDTO.getRoomId());
        return Boolean.TRUE.equals(redisTemplate.delete(cacheKey));
    }

    @Override
    public Long queryOnlinePkObjId(Integer roomId) {
        String cacheKey = cacheBuilder.buildLivingOnlinePkCacheKey(roomId);
        Object pkObjId = redisTemplate.opsForValue().get(cacheKey);
        return pkObjId == null ? null : (Long) pkObjId;
    }
}